package com.itcast.flink.connectors.es;

import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * <p>Description: </p>
 *
 * @author
 * @version 1.0
 * <p>Copyright:Copyright(c)2020</p>
 * @date
 */
public class ElasticSinkApplication {
    
    public static void main(String[] args) throws Exception {
        // 1. 创建运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 2. 读取Socket数据源
        DataStreamSource<String> socketStr = env.socketTextStream("192.168.23.128", 9911, "\n");
        
        //3. 配置ES服务信息,
        List<HttpHost> httpHosts = new ArrayList<>();
        httpHosts.add(new HttpHost("10.10.20.132", 9200, "http"));
        //4. 数据解析处理
        ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>(
                httpHosts,
                new ElasticsearchSinkFunction<String>() {
                    public IndexRequest createIndexRequest(String element) {
                        Map<String, String> json = new HashMap<>();
                        // 解析数据
                        String[] arrValue = String.valueOf(element).split("\t");
                        for (int i = 0; i < arrValue.length; i++) {
                            if (i == 0) {
                                json.put("ip", arrValue[i]);
                            } else if (i == 1) {
                                json.put("time", arrValue[i]);
                            } else if (i == 2) {
                                json.put("type", arrValue[i]);
                            } else if (i == 3) {
                                json.put("api", arrValue[i]);
                            }
                        }
                        return Requests.indexRequest()
                                .index("flink-es")
                                .type("access-log")
                                .source(json);
                    }
                    
                    @Override
                    public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
                        indexer.add(createIndexRequest(element));
                    }
                }
        );
        
        // 5. ES的写入配置
        esSinkBuilder.setBulkFlushMaxActions(1);
        esSinkBuilder.setRestClientFactory(
                restClientBuilder -> {
                    restClientBuilder.setMaxRetryTimeoutMillis(5000);
                }
        );
        
        // 6. 添加ES的写入器
        socketStr.addSink(esSinkBuilder.build());
        socketStr.print().setParallelism(1);
        
        // 7. 执行任务
        env.execute("job");
        
    }
}
